package com.lxy.cloud.mqtt.client.config;

import com.fasterxml.jackson.core.type.TypeReference;
import com.lxy.cloud.mqtt.client.constant.enums.ProcessBeanEnum;
import com.lxy.cloud.mqtt.client.constant.enums.TopicEnum;
import com.lxy.cloud.mqtt.client.model.dto.*;
import com.lxy.cloud.mqtt.client.service.MessageProcessService;
import com.lxy.cloud.mqtt.client.util.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.regex.Pattern;

/**
 * @Author: lxy
 * @Date: 2020/9/6 23:24
 * @Desc: 消费者配置
 */
@Configuration
public class MqttInConfig {

    private static final Logger LOGGER = LoggerFactory.getLogger(MqttConfig.class);

    @Autowired
    private MqttPahoClientFactory mqttClientFactory;

    @Autowired
    private MqttConfig mqttConfig;

    private final Map<String, MessageProcessService> processorMap;

    public MqttInConfig(Map<String, MessageProcessService> processorMap) {
        this.processorMap = processorMap;
    }


    /**
     * MQTT消息订阅绑定（消费者）
     *
     * @return {@link org.springframework.integration.core.MessageProducer}
     */
    @Bean
    public MessageProducer inbound() {
        // 可以同时消费（订阅）多个Topic
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        mqttConfig.getConsumerClientId(), mqttClientFactory,
                        new String[]{TopicEnum.CAR_TO_SERVER.getTopicName(),
                                TopicEnum.REPLY_SERVER_TO_CAR.getTopicName()
                                , TopicEnum.CAR_NOTIFY_SERVER.getTopicName()});
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        // 设置订阅通道
        adapter.setOutputChannel(mqttInboundChannel());
        return adapter;
    }

    /**
     * MQTT信息通道（消费者）
     *
     * @return {@link org.springframework.messaging.MessageChannel}
     */
    @Bean(name = MqttConfig.CHANNEL_NAME_IN)
    public MessageChannel mqttInboundChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT消息处理器（消费者）
     *
     * @return {@link org.springframework.messaging.MessageHandler}
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConfig.CHANNEL_NAME_IN)
    public MessageHandler handler() {
        return message -> {
            LOGGER.info("接收到MQTT消息，消息内容:{}", JsonUtil.create().objectToJson(message));
            String topicName = (String) message.getHeaders().get("mqtt_receivedTopic");
            String payload = message.getPayload().toString();
            payload = payload.replaceAll(" ", "");
            if (Pattern.matches(mqttConfig.getReplyRegex(), topicName)) {
                RelayDTO<RelayDataDTO> relayReq = JsonUtil.create().jsonToObject(payload, new TypeReference<RelayDTO<RelayDataDTO>>() {
                });
                getVinByRegex(topicName, relayReq);
                MessageProcessService messageProcess = processorMap.get(ProcessBeanEnum.COMMON_REPLY.getBeanName());
                messageProcess.onEvent(relayReq);
            } else if (Pattern.matches(mqttConfig.getGetRegex(), topicName)) {
                MqttGetDTO mqttGetReq = JsonUtil.create().jsonToObject(payload, MqttGetDTO.class);
                String method = mqttGetReq.getMethod();
                getVinByRegex(topicName, mqttGetReq);
                MessageProcessService messageProcess = processorMap.get(ProcessBeanEnum.getBeanNameByEventName(method));
                messageProcess.onEvent(mqttGetReq);
            } else if (Pattern.matches(mqttConfig.getEventRegex(), topicName)) {
                MqttEventDTO mqttEventReq = JsonUtil.create().jsonToObject(payload, MqttEventDTO.class);
                String eventName = mqttEventReq.getEventName();
                getVinByRegex(topicName, mqttEventReq);
                MessageProcessService messageProcess = processorMap.get(ProcessBeanEnum.getBeanNameByEventName(eventName));
                messageProcess.onEvent(mqttEventReq);
            }
        };
    }


    private void getVinByRegex(String topic, BaseMqttDTO baseMqttReq) {
        String[] split = topic.split("\\/");
        if (!StringUtils.isEmpty(split[2])) {
            baseMqttReq.setVin(split[2]);
        }
    }


}
